package com.digiwin.demo.sink;

import com.digiwin.demo.util.MongoUtil;
import com.mongodb.client.MongoCollection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.bson.Document;

import java.io.Serializable;

/**
 * @Author yanggld
 * @Date 2019/12/26-14:49
 * @Description
 */
public abstract class AbstractMongoSinkFunction<T> extends RichSinkFunction<T>{
	private String dbName;
	private String colName;
	private static MongoUtil mongoUtil;
	public MongoCollection<Document> collection = null;

	public AbstractMongoSinkFunction(String dbName, String colName) {
		this.dbName = dbName;
		this.colName = colName;
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		mongoUtil = new MongoUtil(dbName,colName);
		collection = mongoUtil.getCollection();
	}

	@Override
	public void close() throws Exception {
		super.close();
		mongoUtil.close();
	}

	@Override
	public abstract void invoke(T value, Context context) throws Exception;
}
